package com.ibm.cps.demo;

import java.io.IOException;

import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.ParseException;
import org.apache.http.client.ClientProtocolException;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.http.util.EntityUtils;

import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.ibm.cps.message.MessageFields;
import com.ibm.interfaces.INoSqlDB;
import com.ibm.mongo.CollectionAccess;
import com.ibm.mongo.MongoNoSqlDB;
import com.ibm.util.JSONUtility;
import com.ibm.util.exception.CPSException;
import com.mongodb.MongoClientURI;

/*
 * this class is to send some graphite message to mongo
 */

class GraphiteData implements Runnable{

	public String subsrcibe_url;
	public String mongo_url;
	private INoSqlDB mongodb;
	private CollectionAccess colAccess;
	private int index;
	
	/*
	 * graphite_url = 9.186.107.31
	 * target = bigant-test-003.cpu-0.cpu-idle
	 */
	public GraphiteData(String graphite_url,String target,String mongo_url) {
		subsrcibe_url = "http://" + graphite_url + "/render/?from=-10s&target="
				+ target + "&format=json";
		this.mongo_url = mongo_url;
		System.out.println(subsrcibe_url);
		System.out.println(mongo_url);
		
		try {
			MongoClientURI uri = new MongoClientURI(mongo_url);
			this.mongodb = new MongoNoSqlDB(mongo_url, uri.getDatabase());
			colAccess = new CollectionAccess(mongodb);
			index = mongodb.getCollection(uri.getCollection());
		} catch (CPSException e) {
			e.printStackTrace();
		}
		
		
	}
	
	
	public String fetch(){
		HttpGet httpGet = new HttpGet(subsrcibe_url);
		CloseableHttpClient httpClient = HttpClientBuilder.create().build();
		String result = null;
		try {
			HttpResponse httpResponse = httpClient.execute(httpGet);
			if (httpResponse.getStatusLine().getStatusCode() == 200) {
				HttpEntity httpEntity = httpResponse.getEntity();
				result = EntityUtils.toString(httpEntity);
				
			}
		} catch (ClientProtocolException e) {
			e.printStackTrace();
		} catch (ParseException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		}
		finally {
			try {
				httpClient.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return result;
	}
	
	public String parseMessage(String result){
		try {
			ObjectNode node = (ObjectNode) JSONUtility.getArrayFromString(result).get(0);
			String target = node.get("target").asText();
			ArrayNode data = (ArrayNode) node.get("datapoints").get(0);
			Long timestamp = data.get(1).asLong();
			Double value = data.get(0).asDouble();
			if (value == null || timestamp == null) {
				return null;
			}
			ObjectNode retNode = JSONUtility.newObjectNode();
			retNode.put(target.replaceAll("[-.]", ""), value);
			retNode.put(MessageFields.TIMESTAMP, timestamp);
			retNode.put(MessageFields.TSKEY, 1);
			return retNode.toString();
		} catch (CPSException e) {
			e.printStackTrace();
		}
		return null;
	}
	
	public void sendToMongo(String result){
		if (result == null) {
			return;
		}
		try {
			ObjectNode doc = JSONUtility.fromObjectString(result);
			System.out.println(doc);
			colAccess.post(index, doc, MessageFields.TIMESTAMP);
		} catch (CPSException e) {
			e.printStackTrace();
		}
	}
	
	@Override
	public void run() {
		while (true) {
			try {
				String result = fetch();
				String object = parseMessage(result);
				sendToMongo(object);
				Thread.currentThread().sleep(10000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}
	
}


public class PublishStaticData {

	public static void main(String[] args) {
		String graphite_url = "9.186.107.31";
		String target = "bigant-test-003.cpu-0.cpu-idle";
		String url = "mongodb://stream-test-007:27017,stream-test-006:27017/graphite.cpu";
		Thread transfer = new Thread(new GraphiteData(graphite_url, target, url));
		transfer.start();
	}

}
